package com.xpec.c4.codec.amf3;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.HashMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.xpec.c4.codec.Codec;

import flex.messaging.io.SerializationContext;
import flex.messaging.io.amf.Amf3Input;
import flex.messaging.io.amf.Amf3Output;

/**
 * 使用AMF方式，进行对象编码与解码
 * 
 * @author lipeilin
 * 
 */
public class Amf3Codec implements Codec {

	private static final Log log = LogFactory.getLog(Amf3Codec.class);

	SerializationContext serializationContext = new SerializationContext();

	ThreadLocal<Amf3Input> amf3Input = new ThreadLocal<Amf3Input>();

	ThreadLocal<Amf3Output> amf3Output = new ThreadLocal<Amf3Output>();

	private static final int CACHE_SIZE = 8000;

	private static final int CACHE_CLEAR_SIZE = (int) (CACHE_SIZE * 0.8);

	/**
	 * amf3的object转换为byte[]的cache
	 */
	private static final HashMap<Object, byte[]> byteArrayCache = new HashMap<Object, byte[]>(
			CACHE_SIZE, 0.9f);

	private static int count = 0;

	/**
	 * 
	 */
	public Amf3Codec() {
		serializationContext.legacyCollection = true;
	}

	private Amf3Input getAmf3Input() {
		// Amf3Input input = new Amf3Input(serializationContext);
		Amf3Input input = amf3Input.get();
		if (input == null) {
			input = new Amf3Input(serializationContext);
			amf3Input.set(input);
		} else {
			input.reset();
		}
		return input;
	}

	private Amf3Output getAmf3Output() {
		// Amf3Output output = new Amf3Output(serializationContext);
		Amf3Output output = amf3Output.get();
		if (output == null) {
			output = new Amf3Output(serializationContext);
			amf3Output.set(output);
		} else {
			output.reset();
		}
		return output;
	}

	@Override
	public Object decode(ByteBuffer byteBuffer) throws Exception {

		byte[] bytes = new byte[byteBuffer.remaining()];
		int beginPosition = byteBuffer.position();
		byteBuffer.get(bytes);

		// decodeByte(bytes);

		// 序列化amf3对象

		InputStream is = new ByteArrayInputStream(bytes);

		Amf3Input amfinput = getAmf3Input();

		amfinput.setInputStream(is);

		int getPosition = amfinput.available();
		Object obj = amfinput.readObject();
		// Object obj1 = amfout.readObject();

		byteBuffer.position(getPosition - amfinput.available() + beginPosition);


		// clear
		// amfinput.reset();
		//amfinput.close();

		return obj;
	}

	@Override
	public ByteBuffer encode(Object obj) throws Exception {

		byte[] messageBytes = encodeByte(obj);

		ByteBuffer buffer = ByteBuffer.allocate(messageBytes.length);
		buffer.put(messageBytes);
		buffer.flip();

		return buffer;
	}
	
	@Override
	public byte[] encodeEx(Object obj) throws Exception {

		byte[] messageBytes = encodeByte(obj);
		return messageBytes;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.codec.Codec#decodeByte(byte[])
	 */
	@Override
	public Object decodeByte(byte[] byteBuffer) throws Exception {

		// 序列化amf3对象

		InputStream is = new ByteArrayInputStream(byteBuffer);

		Amf3Input amfinput = getAmf3Input();

		amfinput.setInputStream(is);

		Object obj = amfinput.readObject();

		// 清空
		// amfinput.reset();
		amfinput.close();
		is.close();

		return obj;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.codec.Codec#encodeByte(java.lang.Object)
	 */
	@Override
	public byte[] encodeByte(Object obj) throws Exception {
		
		byte[] messageBytes = null;
		//byte[] messageBytes = getCache(obj);

		//if (messageBytes == null) {

			// 序列化amf3对象

			// 实现了一个输出流，其中的数据被写入一个 byte 数组。
			ByteArrayOutputStream byteoutStream = new ByteArrayOutputStream();

			// 将byteoutStream产生的数组流导入到DataOutputStream流中
			DataOutputStream dataoutstream = new DataOutputStream(byteoutStream);

			Amf3Output amfout = getAmf3Output();

			// 设置流的编码格式为amf3
			amfout.setOutputStream(dataoutstream);

			amfout.writeObject(obj);// 实际上是将map对象写入到dataoutstream流中
			dataoutstream.flush();// 清空缓存

			// 将ByteArrayOutputStream流中转化成字节数组
			messageBytes = byteoutStream.toByteArray();// amf3数据

			// 清空
			// amfout.reset();
			//byteoutStream.close();
			//amfout.close();
			//dataoutstream.close();

			//putCache(obj, messageBytes);

			// if (count++ > 100) {
			// log.info("Amf3Codec cache size:" + byteArrayCache.size());
			// count = 0;
			// }

		//} else {
//			if (log.isDebugEnabled()) {
//				log.debug("load from cache:" + obj + " cacheSize:"
//						+ byteArrayCache.size());
//			}
		//}

		return messageBytes;
	}

	private void putCache(Object key, byte[] value) {
		if (byteArrayCache.size() > CACHE_CLEAR_SIZE) {
			byteArrayCache.clear();
		}
		byteArrayCache.put(key, value);
	}

	private byte[] getCache(Object key) {
		return byteArrayCache.get(key);
	}

}
